package tss.dsp;

import java.util.*;

import tss.*;

public abstract class Process {
	private Transport transport;
	private int processId;
	private int groupSize;
	private Thread thread;
	private boolean running;
	private Queue<Message> messageQueue;
	private Log log;
	private DSPHelper helper;

	protected Process(int processId, int groupSize, Transport transport, Log log) {
		if (groupSize < 0)
			throw new IllegalArgumentException(
					"Argument groupSize must be greater than zero.");
		if (processId < 0 || groupSize <= processId)
			throw new IllegalArgumentException(
					"Argument processId must satisfy 0 <= processId < groupSize.");
		if (transport == null)
			throw new IllegalArgumentException(
					"Argument transport can not be null.");
		if (log == null)
			throw new IllegalArgumentException("Argument log can not be null.");

		this.processId = processId;
		this.groupSize = groupSize;
		this.transport = transport;
		this.log = log;
		this.messageQueue = new LinkedList<Message>();
		this.helper = new DSPHelper(this, log);

		this.thread = new Thread(new Runnable() {
			public void run() {
				worker();
			};
		});
	}

	private void worker() {
		while (running) {
			try {
				boolean workDone = false;

				Message message = messageQueue.poll();
				if (message != null) {
					transport.send(message);
					workDone = true;
				}

				if (transport.peek() != null) {
					receiveMessageInter(transport.receive());
					workDone = true;
				}

				if (!workDone)
					Thread.sleep(500);
			} catch (InterruptedException ex) {
				synchronized (this) {
					stop();
					log.log(ex);
					return;
				}
			} catch (Exception ex) {
				log.log(ex);
			}
		}
	}

	public int getProcessId() {
		return processId;
	}

	public int getGroupSize() {
		return groupSize;
	}

	public synchronized void start() {
		if (thread == null)
			throw new IllegalStateException(
					"Can not start a process that has been started once.");
		if (running)
			throw new IllegalStateException(
					"Can not start an already running process.");

		running = true;
		thread.start();
	}

	public synchronized void stop() {
		if (thread == null)
			throw new IllegalStateException(
					"Can not stop an already stopped process.");
		if (!running)
			throw new IllegalStateException(
					"Can not stop a process that hasn't been started yet.");

		running = false;
		thread = null;
	}

	protected final void sendMessage(Message message, int... recipients) {
		for (int recipient : recipients)
			message.pushHeader(new RecipientHeader(recipient));
		message.pushHeader(new SenderHeader(getProcessId()));
		helper.sendMessage(message);
		messageQueue.add(message);
	}

	private void receiveMessageInter(Message message) {
		boolean hasRecipients = false, foundRecipient = false;
		int sender = -1;

		for (Object obj : message.getHeaders()) {
			if (!(obj instanceof HeaderItem))
				continue;
			HeaderItem header = (HeaderItem) obj;

			if (header instanceof RecipientHeader) {
				hasRecipients = true;
				if (((RecipientHeader) header).getRecipient() == getProcessId())
					foundRecipient = true;
			}

			if (header instanceof SenderHeader)
				sender = ((SenderHeader) header).getSender();
		}

		if (sender == -1) {
			log.log("Message '" + message.toString()
					+ "' with no sender discarded.");
			return;
		}

		if (!hasRecipients || foundRecipient) {
			helper.receiveMessage(message, sender);
			if (!(message instanceof DSPMessage))
				receiveMessage(message, sender);
		}
	}

	protected abstract void receiveMessage(Message message, int from);

	protected abstract Object getState();

	public void startRecording() {
		helper.startRecording();
	}

	public Snapshot stopRecording() {
		return helper.stopRecording();
	}
}
